#ifndef __MQ_CONNECTION_HPP__
#define __MQ_CONNECTION_HPP__

#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/net/TcpClient.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/base/CountDownLatch.h"
#include "mq_channel.hpp"
#include "mq_worker.hpp"

namespace mq
{
    class Connection
    {
    public:
        using ptr = std::shared_ptr<Connection>;
        Connection(const std::string &sip, int sport, const AsyncWorker::ptr &worker)
            : _latch(1), 
            _client(worker->loopthread.startLoop(), 
            muduo::net::InetAddress(sip, sport), "Client"), 
            _dispatcher(std::bind(&Connection::onUnknownMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)), 
            _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1,                                                                                                                                                                                                                                                                                     std::placeholders::_2, std::placeholders::_3))),
            _worker(worker), _channel_manager(std::make_shared<ChannelManager>())
        {
            _dispatcher.registerMessageCallback<basicCommonResponse>(std::bind(&Connection::basicRespone, this, std::placeholders::_1,
                                                                               std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicConsumeResponse>(std::bind(&Connection::ConsumeResponse, this, std::placeholders::_1,
                                                                               std::placeholders::_2, std::placeholders::_3));
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage,_codec.get(),std::placeholders::_1,
                                                                                std::placeholders::_2,std::placeholders::_3));
            _client.setConnectionCallback(std::bind(&Connection::onConnection,this,std::placeholders::_1));

            _client.connect();
            _latch.wait();
        }

        /*打开信道*/
        Channel::ptr openChannel()
        {
            Channel::ptr channel=_channel_manager->create(_conn,_codec);
            bool ret=channel->openChannel();
            if(ret==false)
            {
                DLOG("打开信道失败");
                return Channel::ptr();
            }
            return channel;
        }

        /*关闭信道*/
        void closeChannel(const Channel::ptr &channel)
        {
            channel->closeChannel();
            _channel_manager->remove(channel->cid());
        }

    private:
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }

        /*正常响应*/
        void basicRespone(const muduo::net::TcpConnectionPtr &conn, const basicCommonResponsePtr &message, muduo::Timestamp)
        {
            //1.找到信道
            Channel::ptr channel=_channel_manager->getChannel(message->cid());
            if(channel.get()==nullptr)
            {
                DLOG("未找到信道");
                return;
            }
            //2.将响应信息添加进信道map中
            channel->putBasicResponse(message);
        }

        /*消费完之后响应*/
        void ConsumeResponse(const muduo::net::TcpConnectionPtr &conn, const basicConsumeResponsePtr &message, muduo::Timestamp)
        {
            //1.找到信道
            Channel::ptr channel=_channel_manager->getChannel(message->cid());
            if(channel.get()==nullptr)
            {
                DLOG("未找到信道");
                return;
            }
            //2.将消费任务封装成任务添加进异步线程池中
            
            _worker->pool.push([channel,message](){channel->consume(message);});
        }
        /*链接成功或链接释放回调*/
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if(conn->connected()==true)
            {
                _conn=conn;
                _latch.countDown();//唤醒主线程的阻塞
                DLOG("链接成功");
            }
            else
            {
                DLOG("链接释放");
                _conn.reset();
            }
        }

    private:
        muduo::CountDownLatch _latch;         // 实现同步
        muduo::net::TcpConnectionPtr _conn;   // 客户端对应的链接
        muduo::net::TcpClient _client;        // 客户端
        ProtobufDispatcher _dispatcher;       // protobuf请求分发器
        ProtobufCodecPtr _codec;              // 协议处理器
        AsyncWorker::ptr _worker;             // 业务线程池句柄
        ChannelManager::ptr _channel_manager; // 信道管理句柄
    };
}

#endif